import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

public class DeleteRalation extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> env, Delete delete, WALEdit edit, Durability durability) throws IOException {
        HTableInterface relationship = env.getEnvironment().getTable(TableName.valueOf("relationship"));
        // 获取删除的rowkey
        byte[] row = delete.getRow();
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();
        for (Map.Entry<byte[], List<Cell>> entry:entries){
            System.out.println("====================" + Bytes.toString(entry.getKey()));
            List<Cell> cells = entry.getValue();
            for (Cell cell :cells){
                byte[] rowkey = CellUtil.cloneRow(cell);
                byte[] column = CellUtil.cloneQualifier(cell);
                System.out.println("rowkey ============= " + rowkey);
                System.out.println("column ============= " + column);
                //判断要删除的是否存在
                boolean flag = relationship.exists(new Get(column).addColumn(Bytes.toBytes("friends"),rowkey));
                if(flag){
                    System.out.println("开始删除 ======================== " + column + "  , " + rowkey );
                    Delete deleteF = new Delete(column).addColumn(Bytes.toBytes("friends"), rowkey);
                    relationship.delete(deleteF);
                }
            }

        }
        relationship.close();

    }
}